package voldemort.store.cachestore.impl;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import voldemort.store.cachestore.CacheBlock;
import voldemort.store.cachestore.Key;

import java.util.TimerTask;

import static voldemort.store.cachestore.impl.CacheStore.State;

/**
 * Created by IntelliJ IDEA.
 * User: mhsieh
 * Date: 3/7/11
 * Time: 12:31 PM
 * To change this template use File | Settings | File Templates.
 */
public class PurgeThread extends TimerTask {
    private static Log logger = LogFactory.getLog(PurgeThread.class);
    private Purge purge;
    private CacheStore store;

    public PurgeThread(Purge purge, CacheStore store) {
        this.purge = purge;
        this.store = store;
    }

    @Override
    public void run() {
        store.getPackLock().lock();
        try {
            store.setServerState(State.Purge);
            int index = store.getCurIndex();
            ChannelStore current =store.getList().get(index);
            //int i = 0;
            //Iterator<Map.Entry<Key, CacheBlock>> iterator = store.getMap().entrySet().iterator();
            logger.info("Start to purge data ...");
            //Iterator <Key> keys = map.
            int totalRecord = current.getTotalRecord();
            int i = 0; int j =0 ; int error = 0;
            long per = totalRecord / 10 ;
            for ( int k=0 ; k < totalRecord ; k++) {
                if ( per > 0 && (k+1) % per == 0 ) {
                    logger.info( (++j * 10) + "% complete deleted "+i+" error "+error);
                }
                try {
                     Key key = current.readKey( k);
                     // key == null means delete flag is on
                     if ( key == null ) {
                         continue;
                     }
                     CacheBlock block = store.getMap().get( key);
                     if ( block == null ) logger.info("key "+key.toString() +" is not in map "+block.toString());
                     else if ( block.getIndex() != index ) {
                         logger.info("skip, index "+block.getIndex() + " <> " + index+ " "+block.toString());
                     }
                     else {
                         synchronized( block) {
                             boolean inFile = false ;
                             byte[] data = (byte[]) block.getData();
                             if ( data == null ) {
                                 inFile = true;
                                 data = current.readChannel( block.getDataOffset2Len(), current.getDataChannel());
                             }
                             try {
                                 if ( purge.isDelete( data, key) ) {
                                     store.remove( key );
                                     i++;
                                 }
                             } catch (RuntimeException rex) {
                                 logger.info(rex.getMessage());
                                 logger.info( "key "+key.toString()+" rec # "+k +" "+inFile + " "+block.toString() );
                                 store.remove( key);
                                 i++;
                             }
//                             if ( block.getData() != null ) {
//                                 if ( purge.isDelete( (byte[]) block.getData(), key)) {
//                                     store.remove( key );
//                                     i++;
//                                 }
//
//                             }
//                             else {
//                                 // read it from channel
//                                 //ChannelStore channel = store.getChannel( entry.getValue());
//                                  data = current.readChannel( block.getDataOffset2Len(), current.getDataChannel());
//                                 //Value value = store.get( entry.getKey() );
//                                 if ( purge.isDelete( data, key) ) {
//                                     store.remove( key );
//                                     i++;
//                                 }
//                             }

                         }
                     }

                 } catch (Exception io) {
                     // swallow exception
                     logger.error(io.getMessage(), io);
                     error ++;

                 }
            }
            logger.info("total records purged "+ i);
        } finally {
            store.setServerState(State.Active);
            store.getPackLock().unlock();
        }
    }



}
